/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.tools;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import org.apache.http.HttpResponse;
import org.apache.http.StatusLine;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.InputStreamEntity;
import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.log4j.Logger;

public class JettyClient {

    /**
     * @param args the command line arguments
     */
    private static Logger logger = Logger.getLogger(JettyClient.class);

    //发给调度用的send方法
    public void send(String schedulerPath, String responseXml, String subjobid) {
        logger.info(subjobid + " " + schedulerPath + " JettyClient send begin!");
        String rs = "";
        try {
            HttpClient httpClient = new DefaultHttpClient();
            HttpPost httppost = new HttpPost(schedulerPath);
            InputStreamEntity reqEntity = new InputStreamEntity(new ByteArrayInputStream(responseXml.getBytes()), -1);
            reqEntity.setContentType("binary/octet-stream");
            reqEntity.setChunked(true);
            httppost.setEntity(reqEntity);
            logger.info(subjobid + " " + schedulerPath + " sending!wait for record!");
            HttpResponse response = httpClient.execute(httppost);
            logger.info(subjobid + " " + schedulerPath + " sending finished!");
            StatusLine st = response.getStatusLine();

            if (st.getStatusCode() == 200) {
                logger.info(subjobid + " " + schedulerPath + " JettyClient send getStatusCode is 200!");
                rs = HttpResponseParser.getResponseContent(response);
            } else {
                logger.info(subjobid + " " + schedulerPath + " JettyClient send getStatusCode is not 200!");
                rs = "";
            }
            httppost.releaseConnection();

        } catch (IOException ex) {
            logger.error(subjobid + " " + schedulerPath + " JettyClient send error!" + ex.getMessage() + "|" + ex.getLocalizedMessage());
        }
        logger.info(subjobid + " " + schedulerPath + " JettyClient send end!");
    }

    //读取ftpCtrl目录下json文件，平均发给各个代理用的send方法
    public String send(String agentIP, String xml, String subjobid, String hdfsPathForRDD, String namelist) {
        logger.debug(subjobid + " " + agentIP + " JettyClient send begin!");
        String rs = "";
        try {
            HttpClient httpClient = new DefaultHttpClient();
            HttpPost httppost = new HttpPost("http://" + agentIP + "/agent/collector");
            //自定义参数这么传过去
            xml = XmlTools.addEl(xml, "subjobid", "agentIPOne", agentIP);
            xml = XmlTools.addEl(xml, "subjobid", "hdfsPathOne", hdfsPathForRDD);

            //添加jsonfileslist节点
            if (namelist.endsWith("|")) {
                namelist = namelist.substring(0, namelist.length() - 1);
            }
            xml = XmlTools.addEl(xml, "subjobid", "jsonfilesname", namelist);

            InputStreamEntity reqEntity = new InputStreamEntity(new ByteArrayInputStream(xml.getBytes()), -1);
            reqEntity.setContentType("binary/octet-stream");
            reqEntity.setChunked(true);
            httppost.setEntity(reqEntity);
            logger.debug(subjobid + " " + agentIP + " sending!wait for record!");
            HttpResponse response = httpClient.execute(httppost);
            logger.debug(subjobid + " " + agentIP + " sending finished!");
            StatusLine st = response.getStatusLine();

            if (st.getStatusCode() == 200) {
                logger.debug(subjobid + " " + agentIP + " JettyClient send getStatusCode is 200!");
                rs = HttpResponseParser.getResponseContent(response);
            } else {
                logger.debug(subjobid + " " + agentIP + " JettyClient send getStatusCode is not 200!");
                rs = "";
            }
            httppost.releaseConnection();

        } catch (IOException ex) {
            logger.debug(subjobid + " " + agentIP + " JettyClient send error!" + ex.getMessage() + "|" + ex.getLocalizedMessage());
            rs = "";
        }
        logger.debug(subjobid + " " + agentIP + " JettyClient send end!");
        return rs;
    }

    //直接把xml发给各个代理用的send方法
    public String send(String agentIP, String xml, String subjobid, String hdfsPathForRDD) {
        logger.info(subjobid + " " + agentIP + " JettyClient send begin!");
        String rs = "";
        try {
            HttpClient httpClient = new DefaultHttpClient();
            HttpPost httppost = new HttpPost("http://" + agentIP + "/agent/collector");
            //自定义参数这么传过去
            xml = XmlTools.addEl(xml, "subjobid", "agentIPOne", agentIP);
            xml = XmlTools.addEl(xml, "subjobid", "hdfsPathOne", hdfsPathForRDD);

            InputStreamEntity reqEntity = new InputStreamEntity(new ByteArrayInputStream(xml.getBytes()), -1);
            reqEntity.setContentType("binary/octet-stream");
            reqEntity.setChunked(true);
            httppost.setEntity(reqEntity);
            logger.info(subjobid + " " + agentIP + " sending!wait for record!");
            HttpResponse response = httpClient.execute(httppost);
            logger.info(subjobid + " " + agentIP + " sending finished!");
            StatusLine st = response.getStatusLine();

            if (st.getStatusCode() == 200) {
                logger.info(subjobid + " " + agentIP + " JettyClient send getStatusCode is 200!");
                rs = HttpResponseParser.getResponseContent(response);
            } else {
                logger.info(subjobid + " " + agentIP + " JettyClient send getStatusCode is not 200!");
                rs = "";
            }
            httppost.releaseConnection();

        } catch (IOException ex) {
            logger.error(subjobid + " " + agentIP + " JettyClient send error!" + ex.getMessage() + "|" + ex.getLocalizedMessage());
            rs = "";
        }
        logger.info(subjobid + " " + agentIP + " JettyClient send end!");
        return rs;
    }

    //ftp平均发给各个代理用的send方法
    public String send(String agentIP, String xml, String subjobid, String hdfsPathForRDD, ArrayList ftplist) {
        logger.info(subjobid + " " + agentIP + " JettyClient send begin!");
        String rs = "";
        try {
            HttpClient httpClient = new DefaultHttpClient();
            HttpPost httppost = new HttpPost("http://" + agentIP + "/agent/collector");
            //自定义参数这么传过去
            xml = XmlTools.addEl(xml, "subjobid", "agentIPOne", agentIP);
            xml = XmlTools.addEl(xml, "subjobid", "hdfsPathOne", hdfsPathForRDD);

            //根节点添加ftplist
            xml = XmlTools.addEl(xml, "subjobid", "ftps", "#####");

            String head = "";
            for (int i = 0; i < ftplist.size(); i++) {
                HashMap<String, String> f = (HashMap<String, String>) ftplist.get(i);
                String ftps = "";
                for (String str : f.keySet()) {
                    ftps += "<" + str + ">" + f.get(str) + "</" + str + ">\n";
                }
                head += "<ftp>\n" + ftps + "</ftp>\n";
            }
            xml = xml.replace("<ftps>#####</ftps>", "<ftps>" + head + "</ftps>");

            InputStreamEntity reqEntity = new InputStreamEntity(new ByteArrayInputStream(xml.getBytes()), -1);
            reqEntity.setContentType("binary/octet-stream");
            reqEntity.setChunked(true);
            httppost.setEntity(reqEntity);
            logger.info(subjobid + " " + agentIP + " sending!wait for record!");
            HttpResponse response = httpClient.execute(httppost);
            logger.info(subjobid + " " + agentIP + " sending finished!");
            StatusLine st = response.getStatusLine();

            if (st.getStatusCode() == 200) {
                logger.info(subjobid + " " + agentIP + " JettyClient send getStatusCode is 200!");
                rs = HttpResponseParser.getResponseContent(response);
            } else {
                logger.info(subjobid + " " + agentIP + " JettyClient send getStatusCode is not 200!");
                rs = "";
            }
            httppost.releaseConnection();

        } catch (IOException ex) {
            logger.error(subjobid + " " + agentIP + " JettyClient send error!" + ex.getMessage() + "|" + ex.getLocalizedMessage());
            rs = "";
        }
        logger.info(subjobid + " " + agentIP + " JettyClient send end!");
        return rs;
    }
}
